package com.poetic.market.analysis.function;

import com.poetic.market.analysis.domain.CountByProvince;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class CountResult extends ProcessWindowFunction<Long, CountByProvince, String, TimeWindow> {
    @Override
    public void process(String key, Context context, Iterable<Long> elements, Collector<CountByProvince> out) throws Exception {

        SimpleDateFormat format = new SimpleDateFormat("yyyy/MM/dd-HH:mm:ss");
        CountByProvince countByProvince = CountByProvince.builder()
                .windowEnd(format.format(new Date(context.window().getEnd())))
                .province(key)
                .count(elements.iterator().next())
                .build();

        out.collect(countByProvince);
    }
}
